package com.guchenbo.example.bigdata;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.RecursiveTask;

/**
 * 固定数量的fork
 *
 * @Author penglai
 * @Date 2016/9/28.
 */
public class ForkTaskMap1 extends RecursiveTask<Map<Integer,List<ListenBean>>> {

    public static final int THRESHOLD = 10000;

    private int start;
    private int end;
    private List<ListenBean> source;

    public ForkTaskMap1(int start, int end) {
        this.start = start;
        this.end = end;
    }

    public ForkTaskMap1(int start, int end, List<ListenBean> source) {
        this.start = start;
        this.end = end;
        this.source = source;
    }

    /**
     * The main computation performed by this task.
     */
    @Override
    protected Map<Integer, List<ListenBean>> compute() {
        Map<Integer, List<ListenBean>> ret = new ConcurrentHashMap<Integer, List<ListenBean>>();
        boolean execute = (end - start) <= THRESHOLD;
        if (execute) {
            for (int i = start; i < end; i++) {
                ListenBean bean = source.get(i);
                Integer key = Integer.valueOf(bean.getTrackId());
                if (!ret.containsKey(key)) {
                    List<ListenBean> list = new ArrayList<ListenBean>();
                    list.add(bean);
                    ret.put(key, list);
                } else {
                    ret.get(key).add(bean);
                }
            }
        } else {
            int middle = (end + start) / 2;
            ForkTaskMap1 leftTask = new ForkTaskMap1(start, middle, source);
            ForkTaskMap1 rightTask = new ForkTaskMap1(middle, end, source);
            // 执行子任务
            leftTask.fork();
            rightTask.fork();

            ret.putAll(leftTask.join());
            ret.putAll(rightTask.join());
        }
        return ret;
    }
}
